/**
 * 
 */
package com.googlecode.xcarve.netty.client;

import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.googlecode.xcarve.common.config.CarveConstances;
import com.googlecode.xcarve.config.ConfigRefer;
import com.googlecode.xcarve.netty.core.DoveFuture;
import com.googlecode.xcarve.object.Header;
import com.googlecode.xcarve.object.NettyMessage;
import com.googlecode.xcarve.object.RPCRequest;
import com.googlecode.xcarve.transport.AbstractClient;

/**
 * @author duxinyun
 *
 */
public class NettyClient extends AbstractClient{
	
	static final Logger logger = LoggerFactory.getLogger(NettyClient.class);
	
	public NettyClient(ConfigRefer config) {
		super(config);
	}
	
	@Override
	public void doConnection() throws RuntimeException {
		ChannelFuture future = null;
		try {
			future = getBootStrap().connect(new InetSocketAddress(config.getHost(),config.getPort()));
			boolean connect = future.awaitUninterruptibly(CarveConstances.DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
			if(connect && future.isSuccess()){
				logger.info(" xcarve establish a new connect successful, address = {}", config.getAddress());
				Channel channel = NettyClient.this.channel;
				try {
					if(channel !=null ){
						channel.close();
					}
				} catch (Exception e) {
				}
				NettyClient.this.channel = future.channel();
				connected();
			}else if(future.cause()!=null){
				logger.error("xcarve rpc create connection fail , address={}, cause={}", new Object[]{config.getAddress(), future.cause()});
			}else if(future.isCancelled()){
				logger.error("xcarve rpc cancell a connection, address = {}", config.getAddress());
			}else{
				logger.error("xcarve rpc create connection timeout, address = {}", config.getAddress());
			}
			 
		} catch (Exception e) {
			logger.warn("xcarve rpc shutdown a connection, address={}, cause={}", new Object[]{config.getAddress(), future.cause()});
			getBootStrap().group().shutdownGracefully();
		}finally{
			if(!isConnection()){
				logger.warn(" this connection is invalid, so cancel it now, address = {}", config.getAddress());
				future.cancel(true);
			}
		}
		
	}

	public void doSend(final Object msg,final long timeout)throws Exception{
		if(getConnection()==null){
			throw new RuntimeException("connection is not avalible.");
		}
//		boolean success = true;
		try {
			final long beginTime = System.currentTimeMillis();
			NettyMessage nettyMsg = new NettyMessage();
			Header header = new Header();
			header.setCodecType((byte)2);
			header.setVersion((byte)1);
			nettyMsg.setHeader(header);
			nettyMsg.setBody(msg);
			ChannelFuture channelFuture = getConnection().writeAndFlush(nettyMsg);
			channelFuture.addListener(new ChannelFutureListener() {
				
				@Override
				public void operationComplete(ChannelFuture future) throws Exception {
					if(future.isSuccess() && msg instanceof RPCRequest){
						RPCRequest request = (RPCRequest)msg;
						if(!request.isHeartBeat()){
							DoveFuture.initSendTime(request.getSeqId());
						}
						return;
					}
					
					if(System.currentTimeMillis() - beginTime > timeout){
						throw new RuntimeException("write message timeout.");
					}
					if(future.isCancelled()){
						logger.info("channel is cancelled by user");
						return;
					}
//					if(!future.isSuccess()){
//						if(channel.isOpen()){
//							channel.close();
//						}
//					}
				}
			});
			
//			success = channelFuture.await(timeout);
//			Throwable cause = channelFuture.cause();
//			if(cause != null){
//				throw cause;
//			}
		} catch (Throwable e) {
			if(e instanceof IOException){
				if(channel != null){
					channel.close();
				}
			}
			throw new RuntimeException("faild to send message.", e.getCause());
		}
//		if(!success){
//			throw new TimeoutException("faild send message to remoteAddress "+ getRemoteAddress() +" in ["+ timeout +"] milis");
//		}
	}
	
	public InetSocketAddress getRemoteAddress(){
		return (InetSocketAddress)channel.remoteAddress();
	}
	
	public InetSocketAddress getLocalAddress(){
		return (InetSocketAddress)channel.localAddress();
	}
}
